/**
 * Epoch.java
 * ************************** 
 * @date Mar 31, 2011
 * @author Michele Tamburini
 * 
 * **************************
 */
package gossipServices.aggregation;

import messages.Message;
import gossipServices.basic.nodeDescriptors.NodeDescriptor;
import gossipServices.gossipMessages.AggregationMessage;


/**
 * Description: 
 *
 */
public class ConcreteEpochHandler implements EpochHandler{
		
	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;
	
	/**
	 * A new node joining the cloud is accepted 
	 * if it can contact a node, belonging to the cloud,
	 * not after that MAXIMUM_CYCLE_DELAY cycles
	 * have elapsed from the beginning of the current epoch.
	 */
	private static final int MAXIMUM_CYCLE_DELAY = 10;
	
	/**
	 * The epoch of the local node.
	 */
	private int epoch = 0;
	
	/**
	 * Number of elapsed cycles from the beginning of the
	 * current epoch.
	 */
	private int elapsedCycles; // delta
	
	/**
	 * An epoch is composed of 
	 * epochCyclesLenght cycles. 
	 */
	private int epochCyclesLenght; //gamma
	
	private int precedingElapsedCycles;
	private int precedingEpoch;

	public final int getEpochCyclesLength() {
		return epochCyclesLenght;
	}

	public final void setEpochCyclesLength(int epochCyclesLength) {
		this.epochCyclesLenght = epochCyclesLength;
	}

	/**
	 * 
	 */
	public ConcreteEpochHandler(int epochCyclesLength) {
		this.epochCyclesLenght = epochCyclesLength;
	}

	/**
	 * @return the epoch
	 */
	@Override
	public int getEpoch() {
		return epoch;
	}
	
	@Override
	public int getRemainingCycles(){
		if(elapsedCycles > MAXIMUM_CYCLE_DELAY )
			return (epochCyclesLenght - elapsedCycles);
		else 
			return 0;
	}

	/**
	 * @return the elapsedCycles
	 */
	@Override
	public int getElapsedCycles() {
		return elapsedCycles;
	}
	
	@Override
	public void elapsedCycle() throws AggregationRestartException{
		elapsedCycles++;
		checkCycles();
	}

	/**
	 * 
	 */
	private void checkCycles() throws AggregationRestartException {
		if(elapsedCycles >= epochCyclesLenght){
			precedingEpoch = epoch;
			precedingElapsedCycles = elapsedCycles;
			increaseEpoch();
			throw new AggregationRestartException(AggregationRestartException.STD_MESSAGE, 
					epoch, 0);
		}
	}

	/**
	 * 
	 */
	private void increaseEpoch() {
		epoch = (epoch + 1);
		elapsedCycles = 0;
	}

	@Override
	public void setEpoch(int newEpoch) {
		this.epoch = newEpoch;
		this.elapsedCycles = 0;
	}

	@Override
	public EpochCheckReturn checkEpochs(AggregationMessage incomingMsg,
			EpochHandler localEpochHandler) {
		EpochCheckReturn ret = null;
		int incomingEpoch = incomingMsg.getActualEpoch();
		int myEpoch = localEpochHandler.getEpoch();
		
//				!incomingMsg.getValue().equals(AggregationMessage.CORRECTION_MESSAGE))
		if(incomingEpoch == myEpoch && 
				incomingMsg.getBody() != null)
			ret = EpochCheckReturn.exchange;
		else if(incomingEpoch < myEpoch)
			ret = EpochCheckReturn.sendCorrectEpoch;
//		else if(incomingMsg.getBody() == null && 
//				incomingEpoch > myEpoch)
		else
			ret = EpochCheckReturn.updateCurrentEpoch;
		return ret; 
	}


	@Override
	public void updateCurrentEpoch(AggregationMessage msgq) throws AggregationRestartException {
		Object body = msgq.getBody();
//		if(correct.getClass().equals(AggregationMessage.CorrectEpoch.class)){
		if(body == null && epoch < msgq.getActualEpoch()){
			precedingElapsedCycles = elapsedCycles;
			precedingEpoch = epoch;
			this.epoch = msgq.getActualEpoch(); 
			elapsedCycles = 0;
			throw new AggregationRestartException(AggregationRestartException.STD_MESSAGE, 
					epoch, msgq.getRemainingCycles());
		}
	}

	@Override
	public Message composeCorrectionMessage(NodeDescriptor me,
			NodeDescriptor sender, Object gossipThread) {
//		Message msg = new AggregationMessage(me, sender, AggregationMessage.CORRECTION_MESSAGE, 
//				epoch, getRemainingCycles());
		Message msg = new AggregationMessage(me, sender, null, gossipThread,
				epoch, getRemainingCycles());
		return msg;
	}

	@Override
	public final int getPrecedingElapsedCycles() {
		return precedingElapsedCycles;
	}

	@Override
	public final int getPrecedingEpoch() {
		return precedingEpoch;
	}


}
